package cn.texous.demo.dmqs.consumer;

import cn.texous.demo.dmqs.executor.AsyncTaskExecutor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

/**
 * kafka 消费者
 *
 * @author leven
 * @since
 */
@Slf4j
@Component
public class KafkaConsumer {

    @Autowired
    private AsyncTaskExecutor asyncTaskExecutor;

    @KafkaListener(topics = "offers", groupId = "offers-consumer")
    public void consumeOffersMsg(ConsumerRecord<String, String> record,
                                 @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {
        String msg = record.value();
        System.out.println(String.format("收到 kafka 消息： %s", msg));
    }

}
